package com.shanzhu.beadhouse.service.kafka;

import com.shanzhu.beadhouse.message.OrderConsumeMessage;
import com.shanzhu.beadhouse.message.OrderTopics;
import lombok.extern.slf4j.Slf4j;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.support.Acknowledgment;
import org.springframework.kafka.support.KafkaHeaders;
import org.springframework.messaging.handler.annotation.Header;
import org.springframework.stereotype.Component;

/**
 * 订单消费记录消息消费者
 */
@Slf4j
@Component
public class OrderConsumeConsumer {

    /**
     * 处理消费记录消息
     * 可以用于后续扩展：发送通知、生成报表等
     */
    @KafkaListener(topics = OrderTopics.ORDER_CONSUME, groupId = "order-consume-group")
    public void handleOrderConsume(OrderConsumeMessage message,
                                   Acknowledgment acknowledgment,
                                   @Header(KafkaHeaders.RECEIVED_TOPIC) String topic,
                                   @Header(KafkaHeaders.RECEIVED_PARTITION_ID) int partition,
                                   @Header(KafkaHeaders.OFFSET) long offset) {
        log.info("收到消费记录消息: topic={}, partition={}, offset={}, message={}",
                topic, partition, offset, message);

        try {
            // 这里可以添加后续的业务逻辑，比如：
            // 1. 发送消费通知给家属
            // 2. 生成消费报表
            // 3. 更新统计数据
            // 4. 触发其他业务流程

            log.info("消费记录消息处理完成: elderId={}, type={}, amount={}",
                    message.getElderId(), message.getConsumeType(), message.getConsumeAmount());

            // 手动确认消息
            acknowledgment.acknowledge();

        } catch (Exception e) {
            log.error("处理消费记录消息失败: elderId={}, type={}",
                    message.getElderId(), message.getConsumeType(), e);
            // 抛出异常，消息会重试
            throw e;
        }
    }
}